package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo3Parallelism {
    public static void main(String[] args) throws Exception {

        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         * flink任务需要的资源和并行度相关，一个并行度对应一个slot
         * 并行度设置方式
         * 1、env.setParallelism(2)： 在创建环境时统一设置，优先级比提交任务时设置要高
         * 2、再提交任务时设置，命令行增加-p参数（推荐使用方式），优先级低
         * 3、每一个算子可以单独设置,优先级最高
         *
         * 并行度设置多少合适 -- 由数据的吞吐量决定（每秒的数据量）
         *
         * 1、非聚合计算时
         * 单个并行度每秒可以处理的数据量在10万-20万左右
         *
         * 2、聚合计算时
         * 每隔并行度每秒处理的数据量在1万-2万左右
         */
        env.setParallelism(2);

        //读取socket
        DataStream<String> lines = env.socketTextStream("master", 8888);

        //一行转换成多行
        DataStream<String> words = lines
                .flatMap((line, collect) -> {
                    String sep = ",";
                    for (String word : line.split(sep)) {
                        collect.collect(word);
                    }
                }, Types.STRING)
                .setParallelism(5);

        //转换成kv格式
        DataStream<Tuple2<String, Integer>> kvs = words
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));

        //按照单词分组
        KeyedStream<Tuple2<String, Integer>, String> keyBy = kvs
                .keyBy(kv -> kv.f0);

        DataStream<Tuple2<String, Integer>> count = keyBy
                .sum(1);

        count.print();

        env.execute();
    }
}
